package com.study.call;

import com.dingtalk.api.DefaultDingTalkClient;
import com.dingtalk.api.DingTalkClient;
import com.dingtalk.api.request.OapiRobotSendRequest;
import com.study.utils.JsonHelper;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.util.Collector;

/**
 * @author sjw
 * @className Check
 * @date 2020/11/28 18:29
 * @description: TODO
 */
public class CallCheck {
    private static DingTalkClient dingTalkClient=new DefaultDingTalkClient("https://oapi.dingtalk.com/robot/send?access_token=3f8134d15c76f137f179102be3d37ef61a7a3b47a65acd759bade3fe4460716f");
    public static void main(String[] args) throws Exception {
        //这些配置建议可以放在配置文件中，然后通过 parameterTool 来获取对应的参数值
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("39.106.17.245").setVirtualHost("OuterHost")
                .setPort(5672).setUserName("root").setPassword("root")
                .build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new RMQSource<>(connectionConfig,
                "q.hiddencall.flink",
                true,
                new SimpleStringSchema()))
                .setParallelism(1)
                .map(new MapFunction<String, CallInfo>() {
                    @Override
                    public CallInfo map(String s) throws Exception {
                        MqMsg mqMsg = JsonHelper.jsonToObject(s, MqMsg.class);
                        if (mqMsg == null) return null;
                        return JsonHelper.jsonToObject(mqMsg.getData(), CallInfo.class);
                    }
                }).keyBy(new KeySelector<CallInfo, String>() {
            @Override
            public String getKey(CallInfo callInfo) throws Exception {
                return callInfo.getCallId();
            }
        }).process(new KeyedProcessFunction<String, CallInfo, String>() {

            private ValueState<CheckInfo> state;

            //最先调用
            @Override
            public void open(Configuration parameters) throws Exception {
                //根据上下文获取状态
                state = getRuntimeContext().getState(new ValueStateDescriptor<CheckInfo>("CheckInfo", CheckInfo.class));
            }

            @Override
            public void processElement(CallInfo value, Context ctx, Collector<String> out) throws Exception {
                if (value.getDuration() > 0 && value.getVoiceUrl() != null) return;
                long ts = System.currentTimeMillis();
                //System.out.println(ts + "   " + value.getCallId());
                CheckInfo checkInfo = state.value();
                if (checkInfo == null) {
                    checkInfo = new CheckInfo();
                    checkInfo.callId = value.getCallId();
                }
                if (value.getDuration() > 0) checkInfo.isDuration = true;
                checkInfo.isVoice = value.getVoiceUrl() != null;
                state.update(checkInfo);
                if(!checkInfo.isDuration || !checkInfo.isVoice) {
                    ctx.timerService().registerProcessingTimeTimer(ts + 20000);
                }else {
                    state.clear();
                }
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                CheckInfo checkInfo = state.value();
                if (checkInfo == null) return;
                if (checkInfo.isDuration && !checkInfo.isVoice)
                    out.collect("业务异常：<" + checkInfo.callId + "> 有通话时长，无通话录音！");
                if (!checkInfo.isDuration && checkInfo.isVoice)
                    out.collect("业务异常：<" + checkInfo.callId + "> 有通话录音，无通话时长！");
                state.clear();
            }
        }).addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                OapiRobotSendRequest request = new OapiRobotSendRequest();
                request.setMsgtype("text");
                OapiRobotSendRequest.Text text = new OapiRobotSendRequest.Text();
                text.setContent(value);
                request.setText(text);
                dingTalkClient.execute(request);
            }
        });
        env.execute("千家电话通话记录业务异常预警");
    }

    private static class CheckInfo {
        private String callId;
        private boolean isDuration;
        private boolean isVoice;
    }
}
